
import org.apache.spark.rdd._
import org.apache.spark.mllib.clustering._
import org.apache.spark.mllib.linalg._
import org.apache.spark.{SparkConf, SparkContext}
object KDD {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("mySpark")
    conf.setMaster("local")
    val sc =new SparkContext(conf)
    val rawData=sc.textFile("/profiledata_06-May-2005/kddcup.data.corrected")

//    rawData.map { line =>
//      val buffer=line.split(',').toBuffer
//      buffer.remove(6,30)
//
//    }
//
//    val a=Array.ofDim[String](6,rawData.count().intValue());

    rawData.map(_.split(',').last).countByValue().toSeq.sortBy(_._2).foreach(println)

    val labelsAndData=rawData.map{line=>
      val buffer=line.split(',').toBuffer
      buffer.remove(1,3)
      val label=buffer.remove(buffer.length-1)
      val vector=Vectors.dense(buffer.map(_.toDouble).toArray)
      (label,vector)
    }

    val data=labelsAndData.values.cache()
    data.saveAsTextFile("data")

    val kmeans=new KMeans()
    kmeans.setK(100)
    kmeans.setEpsilon(1.0e-6)
    val model=kmeans.run(data)
    model.clusterCenters.foreach(println)


    val clusterLabelCount=labelsAndData.map{case(label,datum)=>
      val cluster=model.predict(datum)
      (cluster,label)
    }.countByValue


    clusterLabelCount.toSeq.sorted.foreach{
      case ((cluster,label),count)=>
        println(f"$cluster%1s$label%18s$count%8s")
    }


//    def distance(a:Vector,b:Vector)=
//      math.sqrt(a.toArray.zip(b.toArray).
//        map(p=>p._1-p._2).map(d=>d*d).sum)
//
//    def disTiCentroid(datum:Vector,model: KMeansModel)={
//      val cluster=model.predict(datum)
//      val centroid=model.clusterCenters(cluster)
//      distance(centroid,datum)
//    }
//
//    def clusteringScore(data:RDD[Vector],k:Int)={
//      val kmeans=new KMeans()
//      kmeans.setK(k)
//      val model=kmeans.run(data)
//      data.map(datum=>disTiCentroid(datum,model)).mean()
//    }
//
//    (5 to 80 by 5).map(k=>(k,clusteringScore(data,k))).foreach(println)

  }
}
